package org.raftkv.raft.tracker;

import org.raftkv.raft.exceptions.RaftException;
import org.raftkv.raft.tracker.enums.StateType;

/**
 * 该数据结构用于在 Leader 中保存每个 Follower 的进度。
 * <br/>
 * Leader 将会维护所有 Follower 的进度，并根据其进度向 Follower 发送 entries。
 * <br/>
 * Progress 是一个状态机，其状态变化主要围绕 Raft。此外，某些字段仅在处于特定状态时使用。
 *
 * @author wzy
 */
public class Progress {

    /**
     * match 保存在该节点上保存的日志的最大索引，初始化为 0，正常情况下，next = match + 1。<br/>
     * 以下情况下为非正常情况：
     * <ol>
     *     <li>
     *         切换到 StateProbe 状态时，如果上一个状态是 StateSnapshot，即正在接收快照，
     *         那么 next = Math.max(match + 1, pendingSnapshot + 1)
     *     </li>
     *     <li>
     *         当该 Follower 不在 StateReplicate 状态时，说明不是正常的接收副本状态。
     *         此时当 Leader 与 Follower 同步日志时，可能出现覆盖的情况。<br/>
     *         假设此时 Follower 上面 match 为 3，但是索引为 3 的数据会被 Leader 覆盖，
     *         此时 next 指针可能会一直回溯到与 Leader 上日志匹配的位置再开始正常同步日志。<br/>
     *         此时也会出现 next != match + 1 的情况。
     *     </li>
     * </ol>
     */
    private long match;

    /**
     * next 保存的是下一次 Leader 发送 append 消息时传送过来的日志索引。<br/>
     * 当选举出新的 Leader 时，首先初始化 next 为该 Leader 的最后一条日志 + 1。<br/>
     * 如果向该节点 append 日志失败，则递减 next 回退日志，一直回退到索引匹配为止。
     */
    private long next;

    /**
     * state 定义了 Leader 应该如何与 Follower 互动。<br>
     * <ul>
     *     <li>StateProbe：Leader 在每个心跳的间隔期内至多发送一条 replication 消息，它还会探测 Follower 的实际进度。</li>
     *     <li>
     *         StateReplicate：在发送 replication 消息后，Leader 乐观地增加 next 到最新的 entry。
     *         这是日志条目快速复制到 Follower 的优化状态。
     *     </li>
     *     <li>
     *         StateSnapshot：Leader 在之前应该已经送出了快照，并且停止任何 replication 消息的发送。
     *     </li>
     * </ul>
     */
    private StateType state;

    /**
     * 该字段用于 StateSnapshot 状态。<br/>
     * 如果存在一个挂起的快照，则这个字段的值将被设置为这个快照的索引。<br/>
     * 如果这个字段被设置了，则此 Progress 的复制过程将会被暂停。Raft 不会重新发送快照，直到挂起的快照报告失败。
     */
    private long pendingSnapshot;

    /**
     * 如果该 Progress 最近处于活跃状态，则该字段为 true。<br/>
     * 从相应的 Follower 接收任何消息就表示该 Progress 为活跃状态。<br/>
     * 当一次选举超时的时候，该字段可以被置为 false。
     */
    private boolean recentActive;

    /**
     * 当 Follower 处于 StateProbe 时使用。<br/>
     * 当该字段为 true 时，Raft 应该暂停发送 replication message 到这个 peer 直到重置它。
     *
     * @see Progress#probeAcked
     * @see Progress#isPaused
     */
    private boolean probeSent;

    /**
     * inflights 是一个 inflight message 的滑动窗口。<br/>
     * 每一个 inflight message 包含一个或多个 entry，每一个 message 的最大条目数量在 Raft 中配置（maxSizePerMsg）。<br/>
     * 因此，inflight 有效的限制了 inflight message 和每个 Progress 的带宽。<br/>
     * 当 inflights 已满，不应该再发送任何消息。<br/>
     * 当 Leader 送达一条消息，应该将最后一个 entry 的 index 加入到 inflights 中。<br/>
     * 索引<b style="color: red;">必须</b>按顺序添加到 inflights 中。<br/>
     * 当 Leader 收到回复，前面的 inflights 应该被释放掉，通过调用 freeLE() 方法，并且带上上次收到的 entry 的 index。
     */
    private Inflights inflights;

    /**
     * 如果有一个 Learner 跟踪此 Progress，则 isLearner 为 true
     */
    private boolean isLearner;

    public Progress() {
        this.state = StateType.StateProbe;
    }

    public Progress(long next) {
        this.next = next;
    }


    public Progress(long next, long match) {
        this.next = next;
        this.match = match;
    }

    public void resetState(StateType state) {
        this.probeSent = false;
        this.pendingSnapshot = 0;
        this.state = state;
        this.inflights.reset();
    }

    /**
     * 当 Peer 接受追加时调用此方法。它会重置 probeSent 以指示应立即发送其他追加消息。
     */
    public void probeAcked() {
        this.probeSent = false;
    }

    /**
     * 该函数将 Progress 移动到 StateProbe 状态，并且 next 被重置为 match + 1，
     * 或者，如果更大，则可选地重置为 pendingSnapshot + 1。
     */
    public void becomeProbe() {
        if (this.state == StateType.StateSnapshot) {
            long pendingSnapshot = this.pendingSnapshot;
            resetState(StateType.StateProbe);
            this.next = Math.max(this.match + 1, pendingSnapshot + 1);
        } else {
            resetState(StateType.StateProbe);
            this.next = this.match + 1;
        }
    }

    /**
     * 该函数将 Progress 移动到 StateReplicate 状态，并且重置 next 到 match + 1。
     */
    public void becomeReplicate() {
        resetState(StateType.StateReplicate);
        this.next = this.match + 1;
    }

    /**
     * 该函数将 Progress 移动到 StateSnapshot 状态，并且更新 pendingSnapshot。
     */
    public void becomeSnapshot(long snapIdx) {
        resetState(StateType.StateSnapshot);
        this.pendingSnapshot = snapIdx;
    }

    /**
     * 收到来自 Follower 的 MsgAppResp 后，会调用此方法，来确认索引。
     * <br/>
     * 如果传入的 index 来自过期的消息，则索引不会更新，返回 false；否则更新索引返回 true。
     */
    public boolean maybeUpdate(long index) {
        boolean updated = false;
        if (this.match < index) {
            this.match = index;
            updated = true;
            this.probeAcked();
        }
        this.next = Math.max(this.next, index + 1);
        return updated;
    }

    public void optimisticUpdate(long n) {
        this.next = n + 1;
    }


    /**
     * 该函数把 Progress 调整到拒绝一个 MsgApp 的时候。
     * <br/>
     * 参数是 Follower 拒绝掉的 Message 的索引。
     * <br/>
     * 当消息发送顺序错误或重复时，拒绝可能会虚假发生。在这种情况下，拒绝与进度之前已经确认的索引有关，并且返回 false 而不更改进度。
     * <br/>
     * 如果真的拒绝了，则合理降低 next，并清除 Progress 以发送日志条目。
     */
    public boolean maybeDecrTo(long rejected, long matchHint) {
        if (this.state == StateType.StateReplicate) {
            if (rejected <= this.match) {
                // 这种情况说明返回的情况已经过期，中间有其他添加成功的情况，导致 match 索引递增，此时不需要回退索引，返回 false
                return false;
            }
            this.next = this.match + 1;
            return true;
        }

        // 这种情况说明返回的情况已经过期，不需要回退索引，返回false
        if (next - 1 != rejected) {
            return false;
        }

        // 到了这里就回退 next 为两者的较小值
        this.next = Math.max(Math.min(rejected, matchHint + 1), 1);
        this.probeSent = false;
        return true;
    }

    /**
     * 该函数返回向此节点发送日志条目是否已受到限制。
     * <br/>
     * 当节点拒绝了最近的 MsgApp、当前正在等待快照或已达到 MaxInflightMsgs 的限制时，将执行此操作。
     * <br/>
     * 在正常操作中，将返回 false。
     * <br/>
     * 在达到能够再次接受稳定的日志条目流的状态之前，将不太频繁地联系受限制的节点。
     */
    public boolean isPaused() {
        switch (this.state) {
            case StateProbe:
                return this.probeSent;
            case StateReplicate:
                return this.inflights.full();
            case StateSnapshot:
                return true;
            default:
                throw new RaftException("unexpected state");
        }
    }

    @Override
    public String toString() {
        return "Progress{" +
                "match=" + match +
                ", next=" + next +
                ", state=" + state +
                ", pendingSnapshot=" + pendingSnapshot +
                ", recentActive=" + recentActive +
                ", probeSent=" + probeSent +
                ", inflights=" + inflights +
                ", isLearner=" + isLearner +
                '}';
    }

    @Override
    public Progress clone() {
        Progress p = new Progress();
        p.match = this.match;
        p.next = this.next;
        p.state = this.state;
        p.pendingSnapshot = this.pendingSnapshot;
        p.probeSent = this.probeSent;
        p.inflights = this.inflights.clone();
        p.isLearner = this.isLearner;
        p.recentActive = this.recentActive;
        return p;
    }

    public long getMatch() {
        return match;
    }

    public Inflights getInflights() {
        return inflights;
    }

    public void setInflights(Inflights inflights) {
        this.inflights = inflights;
    }

    public long getNext() {
        return next;
    }

    public StateType getState() {
        return state;
    }

    public void setMatch(long match) {
        this.match = match;
    }

    public void setNext(long next) {
        this.next = next;
    }

    public void setState(StateType state) {
        this.state = state;
    }

    public long getPendingSnapshot() {
        return pendingSnapshot;
    }

    public void setPendingSnapshot(long pendingSnapshot) {
        this.pendingSnapshot = pendingSnapshot;
    }

    public boolean isRecentActive() {
        return recentActive;
    }

    public void setRecentActive(boolean recentActive) {
        this.recentActive = recentActive;
    }

    public boolean isProbeSent() {
        return probeSent;
    }

    public void setProbeSent(boolean probeSent) {
        this.probeSent = probeSent;
    }

    public boolean isLearner() {
        return isLearner;
    }

    public void setLearner(boolean learner) {
        isLearner = learner;
    }

}
